package com.taobao.analyse.reduce;

import java.io.IOException;
import java.util.Iterator;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

import com.taobao.analyse.constants.SplitConstants;

public class AuctionCategoryReducer extends MapReduceBase implements
		Reducer<LongWritable, Text, LongWritable, Text> {

	@Override
	public void reduce(LongWritable key, Iterator<Text> values,
			OutputCollector<LongWritable, Text> output, Reporter reporter)
			throws IOException {
		String curAuctionCatList = null;
		while(values.hasNext()){
			String catList = values.next().toString();
			if(StringUtils.isEmpty(catList) && curAuctionCatList==null){
				//curAuctionCatList = catList;
			}else if(StringUtils.isNotEmpty(catList) && curAuctionCatList==null){
				//第一次进入时将类目列表赋值
				curAuctionCatList = catList;
			}else{
				String[] curList = curAuctionCatList.split(SplitConstants.SPLIT_POUND);
				boolean isChanged = true;
				for(String l : curList){
					String[] list1 = l.split(SplitConstants.SPLIT_UNDER_BAR);
					String[] list2 = catList.split(SplitConstants.SPLIT_UNDER_BAR);
					//过滤掉一些错误的数据
					if(list1.length<1||list2.length<1){
						continue;
					}
					//如果两个一级类目不相等，则在后面添加新数据，如cat1_cat2_cat3_cat4#cat1_cat2_cat3_cat4
					if(list1[0].equals(list2[0])){
						isChanged = false;
					}
				}
				if(isChanged){
					//如果两个一级类目不相等，则在后面添加新数据，如cat1_cat2_cat3_cat4#cat1_cat2_cat3_cat4
					curAuctionCatList+=SplitConstants.SPLIT_POUND + catList;
				}
			}
		}
		output.collect(key, new Text("category@"+curAuctionCatList));
	}
	
	public static void main(String[] args) {
		String abc = "1___";
		System.out.println(abc.split("_").length);
	}

}
